package com.demo.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.demo.rocketmq.config.RocketMQDelayConfig;
import com.demo.rocketmq.constants.DelayLevelEnum;
import com.demo.rocketmq.constants.RocketMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Date;
import java.util.UUID;

/**
 * 自定义时间延迟消息服务
 * 由{@link RocketMQDelayConfig}负责处理延时消息
 * @author cyj
 */
@Slf4j
@Component
public class DelayProducer {

    private final MessageQueueSelector messageQueueSelector = new SelectMessageQueueByHash();

    @Resource
    private RocketMQTemplate rocketMqTemplate;

    /**
     * 发送自定义延迟消息
     * @param topic 主题
     * @param payload 消息
     * @param delaySendTime 消息发送延时时间，单位：秒
     * @return 发送结果
     */
    public SendResult sendDelay(String topic, Object payload, String hashKey, int delaySendTime) {
        log.info("主题:{}，消息：{}，延时时间：{}秒", topic, JSON.toJSONString(payload), delaySendTime);
        Message msg = convertToRocketMessage(topic, payload, hashKey);
        msg.setKeys(hashKey);
        return sendDelay(msg, delaySendTime);
    }

    /**
     * 发送自定义延迟消息
     * @param topic 主题
     * @param payload 消息
     * @param startSendTime 指定消息发送时间
     * @return 发送结果
     */
    public SendResult sendDelay(String topic, Object payload, String hashKey, LocalDateTime startSendTime) {
        log.info("主题:{}，消息：{}，延时时间：{}", topic, JSON.toJSONString(payload), startSendTime);
        long currentTimestamp = LocalDateTime.now().toEpochSecond(ZoneOffset.ofHours(8));
        long futureTimestamp = startSendTime.toEpochSecond(ZoneOffset.ofHours(8));
        long delayTime = futureTimestamp - currentTimestamp;
        if (delayTime <= 0) {
            throw new RuntimeException("startSendTime 必须大于当前时间");
        }
        log.info("延时{}秒后发送消息", delayTime);

        Message msg = convertToRocketMessage(topic, payload, hashKey);
        msg.setKeys(hashKey);
        return sendDelay(msg, Integer.parseInt(delayTime + ""));
    }

    /**
     * 发送自定义延迟消息, 由{@link RocketMQDelayConfig} 创建消息者消费延时消息
     * @param msg 封装好的消息体
     * @param delaySendTime 消息发送延时时间，单位：秒
     * @return 发送结果
     */
    public SendResult sendDelay(Message msg, int delaySendTime) {
        log.info("主题:{}，消息：{}，延时时间：{}秒", msg.getTopic(), JSON.toJSONString(msg.getBody()), delaySendTime);
        try {

            if (delaySendTime <= 0) {
                msg.getProperties().remove(MessageConst.PROPERTY_DELAY_TIME_LEVEL);
                log.info("直接投递, topic={}, tag={}", msg.getTopic(), msg.getTags());
                return sendMsg(msg);
            }

            DelayLevelEnum delayLevel = DelayLevelEnum.calculateDefault(delaySendTime);
            //计算后时间刚好等于延时等级,直接先原主题投递延迟消息
            if (delaySendTime == delayLevel.getTime()) {
                msg.setDelayTimeLevel(delayLevel.getLevel());
                log.info("直接投递, 延时level=1, topic={}, tag={}", msg.getTopic(), msg.getTags());
                return sendMsg(msg);
            }

            fillMessage(msg, delayLevel, delaySendTime);

            return sendMsg(msg);
        } catch (Exception e) {
            log.error("发送消息失败", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 填充消息
     * @param msg 消息体
     * @param delayLevel 延时等级
     * @param delaySendTime 消息发送延时时间，单位：秒
     */
    private void fillMessage(Message msg, DelayLevelEnum delayLevel, int delaySendTime) {
        int newDelaySendTime = delaySendTime - delayLevel.getTime();
        msg.putUserProperty(RocketMqConstants.TIMES, String.valueOf(newDelaySendTime));

        String topic = msg.getProperty(RocketMqConstants.ORIGINAL_TOPIC);
        if (StringUtils.isBlank(topic)) {
            msg.putUserProperty(RocketMqConstants.ORIGINAL_TOPIC, msg.getTopic());
        }
        String tag = msg.getProperty(RocketMqConstants.ORIGINAL_TAG);
        if (StringUtils.isBlank(tag)) {
            if (StringUtils.isNotBlank(msg.getTags())) {
                msg.putUserProperty(RocketMqConstants.ORIGINAL_TAG, msg.getTags());
            }
        }
        String keys = msg.getProperty(RocketMqConstants.ORIGINAL_KEY);
        if (StringUtils.isBlank(keys)) {
            if (StringUtils.isNotBlank(msg.getKeys())) {
                msg.putUserProperty(RocketMqConstants.ORIGINAL_KEY, msg.getKeys());
            }
        }

        String uuid = msg.getProperty(RocketMqConstants.ORIGINAL_UUID);
        if (StringUtils.isBlank(uuid)) {
            uuid = UUID.randomUUID().toString().replaceAll("-", "");
        }

        msg.putUserProperty(RocketMqConstants.ORIGINAL_UUID, uuid);
        msg.setDelayTimeLevel(delayLevel.getLevel());
        msg.setTopic(RocketMqConstants.PROXY_TOPIC);
        log.info("消息uuid {} 发送时间为{},延时等级本次建议为{},延时时间为{}秒, 实际延时{}秒后业务系统收到信息", uuid, String.format("%tF %<tT",
                new Date()), delayLevel.getLevel(), delayLevel.getTime(), newDelaySendTime);
    }

    private SendResult sendMsg(Message msg) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        DefaultMQProducer producer = rocketMqTemplate.getProducer();
        String hashKey = msg.getKeys();
        SendResult sendResult;
        if (StringUtils.isBlank(hashKey)) {
            sendResult = producer.send(msg);
        } else {
            sendResult = producer.send(msg, messageQueueSelector, hashKey);
        }
        logSendResult(sendResult);
        return sendResult;
    }

    private org.apache.rocketmq.common.message.Message convertToRocketMessage(String topic, Object payload, String hashKey) {

        byte[] bytes = JSON.toJSONBytes(payload);
        org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(topic, bytes);
        if (StringUtils.isNotBlank(hashKey)) {
            msg.setKeys(hashKey);
        }
        return msg;
    }

    private void logSendResult(SendResult sendResult) {
        log.info("发送消息的msgId：{}， 发送状态：{}", sendResult.getMsgId(), sendResult.getSendStatus());
    }
}
